package com.run.go.gearbox.core.gearbox.disruptor;


import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import com.run.go.gearbox.core.gearbox.gear.AbstractGear;
import com.run.go.gearbox.core.gearbox.gear.GearTooth;
import com.run.go.gearbox.core.transmission.ExecuteEntity;
import com.run.go.gearbox.core.thread.NamedThreadFactory;



/**
 * disruptor的档位
 */
public class DisruptorGear extends AbstractGear {

    private Disruptor<ExecuteEntity> disruptor;

    private RingBuffer<ExecuteEntity> ringBuffer;

    private WorkHandler<ExecuteEntity> [] handlers;

    public DisruptorGear(Integer power){

        super(power);
        handlers = new WorkHandler[power];

        /* 齿轮队列 */
        this.disruptor = new Disruptor(new ExecuteFactory() ,power, new NamedThreadFactory("disruptor-gear"),ProducerType.MULTI, new BlockingWaitStrategy());
        this.disruptor.setDefaultExceptionHandler(new DisruptorException());
        this.disruptor.handleEventsWithWorkerPool(initHandlers());

    }

    @Override
    public void gear() {
        this.ringBuffer = disruptor.start();
    }

    @Override
    public void downGear() {
        disruptor.shutdown();
    }

    @Override
    public int surplusPower() {
        return (int) ringBuffer.remainingCapacity();
    }

    private WorkHandler<ExecuteEntity> [] initHandlers() {
        for (int i = 0; i < handlers.length; i++) {
            GearTooth gearTooth = new GearTooth();
            handlers[i] = new ExecuteWorkHandler(gearTooth);
        }
        return handlers;
    }

    @Override
    public void execute(ExecuteEntity executeEntity) {
        ringBuffer.publishEvent((executeEntity1,seq) ->{
            executeEntity1.setExecuteId(String.valueOf(seq));
            executeEntity1.setData(executeEntity.getData());
            executeEntity1.setFuture(executeEntity.getFuture());
            executeEntity1.setSqlType(executeEntity.getSqlType());
            executeEntity1.setSql(executeEntity.getSql());
        });
    }


    private static class ExecuteWorkHandler implements WorkHandler<ExecuteEntity>{

        private GearTooth gearTooth;

        private ExecuteWorkHandler(GearTooth gearTooth){
            this.gearTooth = gearTooth;
        }

        @Override
        public void onEvent(ExecuteEntity executeEntity) throws Exception {
            gearTooth.request(executeEntity);
        }
    }

}
